/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.msg;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client. v1.session.RESendSessionBuilder;
import cn.ac.iie.di.gloabl.GlobalParas;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.log4j.Logger;

/**
 *
 * @author lenovo-pc
 */
public class JsonMsgSender implements Runnable {

	private int addCount = 0;
	private long startTime = System.currentTimeMillis();
	static private final Logger logger = Logger.getLogger(JsonMsgSender.class.getName());

	ThreadPoolExecutor threadpool;

	public JsonMsgSender() {
		//创建一个线程池
		threadpool = new ThreadPoolExecutor(12, 12, 3, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(GlobalParas.POOL_MSG),new ThreadFactory() {
                //threadpool = new ThreadPoolExecutor(12, 12, 3, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(20),new ThreadFactory() {
				AtomicInteger idx = new AtomicInteger(0);

				@Override
				public Thread newThread(Runnable r) {
					return new Thread(r, "JsonMsgTask_" + idx.getAndIncrement());
				}
			}, new RejectedExecutionHandler() {

				@Override
				public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
					while (true) {
						try {
							executor.getQueue().put(r);
							break;
						} catch (InterruptedException ex) {
							logger.warn("", ex);
						}
					}
				}
			});
	}

	@Override
	public void run() {

		RESendSession session;
		ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
		ReentrantReadWriteLock.ReadLock readlock = readWriteLock.readLock();
		ReentrantReadWriteLock.WriteLock writelock = readWriteLock.writeLock();
		SessionPool pool;
		try {
			REConnection conn = new REConnection(GlobalParas.IP_PORT);
                       // REConnection conn = new REConnection("10.136.64.28:9877");
			//builder
			RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder(GlobalParas.MQ_MSG);
                        //RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder("qinjia_mq");
			//群发的类型初始化
                        for (String columntype : GlobalParas.TYPE_MSG.split("\\|")) {
                         	switch (columntype.split("[:]")[1]) {
					case "int":
						builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Int, true);
						break;
					case "long":
						builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Long, true);
						break;
					case "string":
						builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.String, true);
						break;
					case "strings":
						builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Strings, true);
						break;
                                        case "boolean":
                                                builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Boolean, true);
					case "struct":
						HashMap<String, REFieldType> detailMap1 = new HashMap<>();
						detailMap1.put("tct_l1", REFieldType.Long);
						detailMap1.put("tct_l2", REFieldType.Long);
						detailMap1.put("tct_l3", REFieldType.Long);
						builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Struct, true);
						REFieldType.Struct.setDetail(columntype.split("[:]")[0].toLowerCase(), detailMap1);
						break;
					case "structs":
						HashMap<String, REFieldType> detailMap2 = new HashMap<>();
						detailMap2.put("v", REFieldType.String);
						detailMap2.put("w", REFieldType.Long);
						builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Structs, true);
						REFieldType.Structs.setDetail(columntype.split("[:]")[0].toLowerCase(), detailMap2);
						break;
					default:
						logger.error("The type is not found");
				}
			}
			pool = new SessionPool(builder);
			//session
			session = (RESendSession) builder.build();

			session.start();

		} catch (Exception ex) {
			logger.error("init session " + ex, ex);
			return;
		}

		String json = null;
		ArrayList<String> jsonList;
		while (true) {
			jsonList = new ArrayList<>(50);
			try {

				for (int i = 0; i < 50; ++i) {
					String str = GlobalParas.msgQueue.poll(500, TimeUnit.MILLISECONDS);
					if (str == null) {
						break;
					}
					jsonList.add(str);
				}
				if (jsonList.isEmpty()) {
					Thread.sleep(5000);
					continue;
				}
				threadpool.execute(new JsonMsgTask1(jsonList, pool));
				++addCount;
			} catch (Exception ex) {
				logger.error("this group thread error :" + ex, ex);
				ex.printStackTrace();
			}
		}

	}

	static public class SessionPool {

		RESendSessionBuilder builder;
		LinkedList<RESendSession> pool;

		public SessionPool(RESendSessionBuilder builder) {
			this.builder = builder;
			pool = new LinkedList<>();
		}

		public synchronized RESendSession borrowSession() throws RESessionException, REConnectionException {
			try {
				return pool.removeFirst();
			} catch (NoSuchElementException ex) {
				RESendSession rtn = (RESendSession) builder.build();
				rtn.start();
				return rtn;
			}
		}

		public synchronized void returnSession(RESendSession session) {
			pool.add(session);
		}
	}
	
}
